package com.blue.user.mq.config;

import com.blue.user.mq.beans.QueueListenerMapEnum;
import com.blue.user.mq.beans.RabbitBeanUtil;
import com.blue.user.mq.beans.RabbitConnConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Configuration;

import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @author liulei
 * @version 1.0
 */
@Slf4j
@Configuration
@ConditionalOnProperty(value = "spring.rmq.enable")
@RequiredArgsConstructor
public class MultiRabbitMqConfig implements InitializingBean {

    @Autowired
    private RabbitBeanUtil beanUtils;

    @Autowired
    private RabbitConnConfig connConfig;

    /**
     * 填充MQ配置
     */
    private Map<String, RabbitConnConfig.Config> enrichRabbitMqConfig() {
        Map<String, RabbitConnConfig.Config> multiMqPropertiesMap = new LinkedHashMap<>(8);
        for (RabbitConnConfig.Config config : connConfig.getConfigs()) {
            RabbitConnConfig.ENV.add(config.getEnv());
            multiMqPropertiesMap.put(config.getEnv(), RabbitConnConfig.Config.builder()
                    .addresses(config.getAddresses())
                    .username(config.getUsername())
                    .password(config.getPassword())
                    .virtualHost(config.getVirtualHost())
                    .build());
        }
        return multiMqPropertiesMap;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Map<String, RabbitConnConfig.Config> multiMqPropertiesMap = this.enrichRabbitMqConfig();
        DefaultListableBeanFactory defaultListableBeanFactory = beanUtils.getDefaultListableBeanFactory();
        AutowireCapableBeanFactory autowireCapableBeanFactory = beanUtils.getAutowireCapableBeanFactory();
        AtomicReference<Integer> envIndex = new AtomicReference<>(0);
        multiMqPropertiesMap.forEach((key, rabbitProperties) -> {
            AbstractBeanDefinition beanDefinition = BeanDefinitionBuilder.genericBeanDefinition(CachingConnectionFactory.class)
                    .addPropertyValue("cacheMode", CachingConnectionFactory.CacheMode.CHANNEL)
                    .addPropertyValue("addresses", rabbitProperties.getAddresses())
                    .addPropertyValue("username", rabbitProperties.getUsername())
                    .addPropertyValue("password", rabbitProperties.getPassword())
                    .addPropertyValue("virtualHost", rabbitProperties.getVirtualHost())
                    .getBeanDefinition();
            String connectionFactoryName = String.format("%s%s", key, "ConnectionFactory");
            defaultListableBeanFactory.registerBeanDefinition(connectionFactoryName, beanDefinition);
            CachingConnectionFactory connectionFactory = defaultListableBeanFactory.getBean(connectionFactoryName, CachingConnectionFactory.class);

            String rabbitAdminName = String.format("%s%s", key, "RabbitAdmin");
            AbstractBeanDefinition rabbitAdminBeanDefinition = BeanDefinitionBuilder.genericBeanDefinition(RabbitAdmin.class)
                    .addConstructorArgValue(connectionFactory)
                    .addPropertyValue("autoStartup", true)
                    .getBeanDefinition();
            defaultListableBeanFactory.registerBeanDefinition(rabbitAdminName, rabbitAdminBeanDefinition);
            RabbitAdmin rabbitAdmin = defaultListableBeanFactory.getBean(rabbitAdminName, RabbitAdmin.class);
            log.info("rabbitAdmin:[{}]", rabbitAdmin);

            RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
            defaultListableBeanFactory.registerSingleton(String.format("%s%s", key, "RabbitTemplate"), rabbitTemplate);

            // 初始化对应env的queue的监听器
            List<String> monitorQueues = QueueListenerMapEnum.getMonitorQueuesByEnv(Integer.valueOf(envIndex.toString()));
            monitorQueues.forEach(queue -> {
                try {
                    SimpleMessageListenerContainer simpleMessageListenerContainer = this.initListenerContainer(connectionFactory, Integer.valueOf(envIndex.toString()), queue);
                    String containerName = String.format("%s%s", key, simpleMessageListenerContainer.getMessageListener().getClass().getSimpleName());
                    defaultListableBeanFactory.registerSingleton(containerName, autowireCapableBeanFactory.initializeBean(simpleMessageListenerContainer, containerName));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
            envIndex.getAndSet(envIndex.get() + 1);
        });
    }

    /**
     * 动态初始化对应MQ的监听容器
     */
    private SimpleMessageListenerContainer initListenerContainer(CachingConnectionFactory connectionFactory, Integer envIndex, String queueName) throws Exception {
        SimpleMessageListenerContainer simpleMessageListenerContainer = this.initListenerBaseConfig(connectionFactory, queueName);
        // 设置消息监听器
        QueueListenerMapEnum monitorEnum = QueueListenerMapEnum.getClassByEnvAndQueue(envIndex, queueName);
        DefaultListableBeanFactory defaultListableBeanFactory = beanUtils.getDefaultListableBeanFactory();
        MessageListener factoryBean = (MessageListener) defaultListableBeanFactory.getBean(monitorEnum.getListener());
        simpleMessageListenerContainer.setMessageListener(factoryBean);
        return simpleMessageListenerContainer;
    }

    /**
     * 统一初始化MQ监听器基础配置
     */
    private SimpleMessageListenerContainer initListenerBaseConfig(CachingConnectionFactory connectionFactory, String queueName) {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        // 设置默认监听的队列
        simpleMessageListenerContainer.setQueueNames(queueName);
        // 指定要创建的并发使用者的数量,默认值是1,当并发高时可以增加这个的数值，同时下方max的数值也要增加
        simpleMessageListenerContainer.setConcurrentConsumers(3);
        // 最大的并发消费者
        simpleMessageListenerContainer.setMaxConcurrentConsumers(10);
        // 设置是否重回队列
        simpleMessageListenerContainer.setDefaultRequeueRejected(false);
        // 设置签收模式
        simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置非独占模式
        simpleMessageListenerContainer.setExclusive(false);
        // 设置consumer未被 ack 的消息个数
        simpleMessageListenerContainer.setPrefetchCount(1);
        return simpleMessageListenerContainer;
    }
}
